package com.csdj.middleware;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;
import java.util.stream.Collectors;

public class RocketMqProducer {

    private TransactionMQProducer producer;
    //事务消息监听器
//    private TransactionListener transactionListener;

    //默认主题
    private String defaultTopic="test";

    public void setDefaultTopic(String topic){
        this.defaultTopic = topic;
    }


    public RocketMqProducer(String producerGroup,String nameHost,TransactionListener transactionListener){
        //创建消息生产者
        producer = new TransactionMQProducer(producerGroup);
        //nameserve 可以是以集群的方式运行， 配置 host1:port;host2:port;host3:port
        producer.setNamesrvAddr(nameHost);//设置nameserver
        producer.setTransactionListener(transactionListener);//设置事务消息监听器
        try {
            producer.start();//启动生产者
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        //添加程序关闭钩子，当程序关闭时也关闭生产者
        Runtime.getRuntime().addShutdownHook(new Thread(()->{
            producer.shutdown();
        }));
    }


    public SendResult sendMsg(String msg){
        return sendMsg(defaultTopic,null,msg);
    }

    /**
     * 发送同步消息
     * @param topic
     * @param tags
     * @param msg
     * @return
     */
    public SendResult sendMsg(String topic,String tags, String msg){
        final Message message = new Message(topic,tags, msg.getBytes());
        try {
            return producer.send(message);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 发送异步消息
     * @param topic
     * @param tags
     * @param msg
     * @param callback
     */
    public void sendMsg(String topic, String tags, String msg, SendCallback callback){
        final Message message = new Message(topic,tags, msg.getBytes());
        try {
            producer.send(message,callback);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送单向消息
     * @param topic
     * @param tags
     * @param msg
     */
    public void sendMsgOneway(String topic, String tags, String msg){
        final Message message = new Message(topic,tags, msg.getBytes());
        try {
            producer.sendOneway(message);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送顺序消息
     * @param topic
     * @param tags
     * @param msg
     * @param orderHash 选择消息队列的标识
     */
    public SendResult sendMsgByOrder(String topic, String tags, String msg,long orderHash){
        final Message message = new Message(topic,tags, msg.getBytes());
        try {
            return producer.send(message, new MessageQueueSelector() {
                //消息队列选择器
                /**
                 *
                 * @param mqs 同一个topic的消息队列集合
                 * @param msg 消息消息内容
                 * @param arg 选择消息队列标识
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    long hash = (long)arg;
                    int index = (int)hash%mqs.size();
                    return mqs.get(index);
                }
            },orderHash);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 发送延迟消息
     * @param topic
     * @param tags
     * @param msg
     * @param delayLevel 延迟等级
     * @return
     */
    public SendResult sendDelay(String topic, String tags, String msg,int delayLevel){
        final Message message = new Message(topic,tags, msg.getBytes());
        message.setDelayTimeLevel(delayLevel);
        try {
            return producer.send(message);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 批量发送消息
     * @param topic
     * @param tags
     * @param msgs
     * @return
     */
    public SendResult sendMsgBatch(String topic,String tags,List<String> msgs){
        final List<Message> list = msgs.stream().map(msg -> {
            return new Message(topic, tags, msg.getBytes());
        }).collect(Collectors.toList());
        try {
            return producer.send(list);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
     * 发送事务消息
     * @param topic
     * @param tags
     * @param msg
     * @return
     */
    public SendResult sendTranscationMsg(String topic,String tags,String msg,Object args){
        final Message message = new Message(topic,tags, msg.getBytes());
        try {
            return producer.sendMessageInTransaction(message,args);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return null;
    }


}
